"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : security-CmdInjection
Case Name   : 查看 gs_ctl 工具中 restore 命令是否存在命令行注入
Create At   : 2022/12/15 20:05
Owner       : x00454388
Description :
    gs_ctl restore 命令中会将用户输入的路径参数格式化打印构造一个 tar 命令:
        snprintf_s(tar_cmd, MAXPGPATH, MAX_PATH_LEN - 1,
         "tar -zvxf %s/base.tar.gz -C %s --strip-components 1"...);
        然后调用 shell 执行。输出控制符 %s 没加引号，
        而 Linux 下文件名可以为非路径分隔符 '/' 外的任意字符，可能存在命令行注入问题。
    1. 以 omm 用户 SSH 登录数据库服务器，执行如下命令：
        cd /tmp
        touch gs_ctl_test
        tar -zcf base.tar.gz gs_ctl_test
        mkdir "/tmp/base.tar.gz --to-command id #"
        cp base.tar.gz "/tmp/base.tar.gz --to-command id #"
    2. 执行 gs_ctl restore --obsmode -D "/tmp/base.tar.gz --to-command id #"
Expect      :
    1. 执行成功
    2. 注入的命令 'id' 没有被执行
History     :
"""


import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger


class CmdInjection(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f'----{os.path.basename(__file__)} start----')
        self.userNode = Node('PrimaryDbUser')
        self.DB_ENV_PATH = macro.DB_ENV_PATH
        self.dir_name = 'base.tar.gz --to-command id #'

    def test_security(self):
        excute_cmd1 = f'''cd /tmp;
                          touch gs_ctl_test;
                          tar -zcf base.tar.gz gs_ctl_test;
                          mkdir "{self.dir_name}";
                          cp base.tar.gz "./{self.dir_name}"
                       '''

        excute_cmd2 = f'source {self.DB_ENV_PATH};gs_ctl --help'
                               
        excute_cmd3 = f'''source {self.DB_ENV_PATH};
                          gs_ctl restore --obsmode -D "/tmp/{self.dir_name}"
                       '''        
        self.userNode.sh(excute_cmd1)
        return_code = self.userNode.sh(excute_cmd2).return_code()
        self.logger.info("return code: " + str(return_code))
        self.assertTrue(return_code == 0)
        msg = self.userNode.sh(excute_cmd3).result()
        self.logger.info(msg)
        self.assertTrue(msg.find(' groups=') == -1 and msg.find(' 组=') == -1)

    def tearDown(self):
        excute_cmd = f'''cd /tmp;
                         rm -f gs_ctl_test;
                         rm -f base.tar.gz;
                         rm -fR "{self.dir_name}"
                      '''
        self.userNode.sh(excute_cmd)
        self.logger.info(f'----{os.path.basename(__file__)} end----')